package com.atguigu.wordCount;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author gmd
 * @desc DataStream实现WordCount（socket无界流+禁用算子链）
 * @since 2024-11-23 11:09:26
 */
public class OperatorChain {

    public static void main(String[] args) throws Exception {
        // IDEA运行时，也可以看到webui，一般用于本地测试
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        // 在idea运行，不指定并行度，默认就是 电脑的 线程数
        env.setParallelism(1);
        // 全局禁用 算子链
        env.disableOperatorChaining();

        // 从socket无界流中读取数据
        DataStreamSource<String> socketDS = env.socketTextStream("127.0.0.1", 7777);

        // 处理数据: 切换、转换、分组、聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = socketDS
                // 禁用算子链化，防止当前算子与前后算子链化在一起
                // .disableChaining()
                .flatMap((String value, Collector<String> out) -> {
                            String[] words = value.split(" ");
                            for (String word : words) {
                                out.collect(word);
                            }
                        }
                )
                // 开启新的算子链
                .startNewChain()
                // 禁用算子链化，防止当前算子与前后算子链化在一起
                // .disableChaining()
                .returns(Types.STRING)
                .map(word -> Tuple2.of(word, 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(value -> value.f0)
                .sum(1);


        // 输出
        sum.print();
        // 执行
        env.execute();
    }

}

/*
 * 1. 算子之间的传输关系
 * 一对一 (One-to-One)：数据按照原样传递到下一个算子，不涉及重新分区。
 * 重分区 (Repartition)：数据会在不同的算子之间重新分区，从而打破原有的数据流动路径。
 *
 * 2. 算子链化的条件
 * 一对一：数据流必须是One-to-One的关系。
 * 并行度相同：算子的并行度（Parallelism）必须相同。
 *
 * 3. 关于算子链的API操作
 * 全局禁用算子链：使用env.disableOperatorChaining()，会全局禁用Flink中的算子链化。
 * 禁用某个算子的链化：使用算子A.disableChaining()，使算子A既不与前面的算子链化，也不与后面的算子链化。
 * 从某个算子开启新链条：使用算子A.startNewChain()，算子A不会与前面的算子链化，但会从它开始进行新的链化。
 */